package com.example.mq.mqserver;

import com.example.mq.common.Consumer;
import com.example.mq.common.MqException;
import com.example.mq.mqserver.core.*;
import com.example.mq.mqserver.datacenter.DiskDataCenter;
import com.example.mq.mqserver.datacenter.MemoryDataCenter;

import java.io.IOException;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;


/**
 * @Description: 虚拟主机
 * 每个虚拟主机下面都管着交换机、队列、绑定、消息数据
 * 同时提供api供上层使用
 * @Author: liudong
 * @Time: 2024/5/16
 */

public class VirtualHost {
    private String virtualHostName;
    private MemoryDataCenter memoryDataCenter = new MemoryDataCenter();
    private DiskDataCenter diskDataCenter = new DiskDataCenter();
    private Router router = new Router();
    private ConsumerManager consumerManager = new ConsumerManager(this);
    //操作交换机的锁对象
    private final Object exchangeLocker = new Object();
    //操作队列的锁对象
    private final Object queueLocker = new Object();

    public String getVirtualHostName() {
        return virtualHostName;
    }

    public MemoryDataCenter getMemoryDataCenter() {
        return memoryDataCenter;
    }

    public DiskDataCenter getDiskDataCenter() {
        return diskDataCenter;
    }


    //使用构造方法来初始化
    public VirtualHost(String name) {
        this.virtualHostName = name;
        //memoryDataCenter 不需要额外的初始化，只要对象new出来就行
        //diskDataCenter 需要初始化，建库建表，初始化数据的设定
        diskDataCenter.init();

        //进行硬盘中的内容，恢复到内存中
        try {
            memoryDataCenter.recovery(diskDataCenter);
        } catch (IOException | MqException | ClassNotFoundException e) {
            e.printStackTrace();
            System.out.println("[VirtualHost] 恢复内存数据失败！");
        }
    }


    /**
     * 交换机操作
     *
     * @param exchangeName
     * @param exchangeType
     * @param durable
     * @param autoDelete
     * @param arguments
     * @return
     */
    //创建交换机
    //交换机不存在就创建，如果存在直接返回
    //创建成功，返回true，失败返回false
    public boolean exchangeDeclare(String exchangeName, ExchangeType exchangeType, boolean durable, boolean autoDelete, Map<String, Object> arguments) {
        //把交换机的名字加上虚拟主机的名字作为前缀
        exchangeName = virtualHostName + exchangeName;
        try {
            //这里和删除交换机都进行加锁，为了防止多个线程操作，保证线程安全
            synchronized (exchangeLocker) {
                //1.判断该交换机是否已经存在，直接通过内存查询
                Exchange existsExchange = memoryDataCenter.getExchange(exchangeName);
                if (existsExchange != null) {
                    //交换已经存在
                    System.out.println("[VirtualHost] 交换机已经存在！ exchangeName = " + exchangeName);
                    return true;
                }
                //2.真正创建交换机，先构造Exchange对象
                Exchange exchange = new Exchange();
                exchange.setName(exchangeName);
                exchange.setType(exchangeType);
                exchange.setDurable(durable);
                exchange.setAutoDelete(autoDelete);
                exchange.setArguments(arguments);
                //3.把交换机对象写入硬盘
                if (durable) {
                    diskDataCenter.insertExchange(exchange);
                }
                //4.把交换机对象写入内存
                memoryDataCenter.insertExchange(exchange);
                System.out.println("[VirtualHost] 交换机创建完成！ exchangeName = " + exchangeName);
                //先写硬盘，是因为硬盘容易写失败，硬盘写失败，内存就不写了
                //先写内存，内存写成功了，硬盘写失败了，还需要把内存的数据再给删除掉，就比较麻烦
            }
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }

    //删除交换机
    public boolean exchangeDelete(String exchangeName) {
        exchangeName = virtualHostName + exchangeName;
        try {
            synchronized (exchangeLocker) {
                //1.先找到对应的交换机
                Exchange toDelete = memoryDataCenter.getExchange(exchangeName);
                if (toDelete == null) {
                    throw new MqException("[VirtualHost] 交换机不存在！无法删除");
                }
                //2.删除硬盘上的数据
                if (toDelete.isDurable()) {
                    diskDataCenter.deleteExchange(exchangeName);
                }
                //3.删除内存上的数据
                memoryDataCenter.deleteExchange(exchangeName);
                System.out.println("[VirtualHost] 交换机删除成功！");
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 交换机删除失败！ exchangeName = " + exchangeName);
            e.printStackTrace();
            return false;
        }
    }


    /**
     * 队列操作
     *
     * @param queueName
     * @param durable
     * @param exclusive
     * @param autoDelete
     * @param arguments
     * @return
     */
    //创建队列
    public boolean queueDeclare(String queueName, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments) {
        //把队列的名字，给拼接上虚拟主机的名字
        queueName = virtualHostName + queueName;
        try {
            synchronized (queueLocker) {
                //1.判断队列是否存在
                MSGQueue existsQueue = memoryDataCenter.getQueue(queueName);
                if (existsQueue != null) {
                    System.out.println("[VirtualHost] 队列已经存在！queueName=" + queueName);
                    return true;
                }
                //2.创建队列对象
                MSGQueue queue = new MSGQueue();
                queue.setName(queueName);
                queue.setDurable(durable);
                queue.setExclusive(exclusive);
                queue.setAutoDelete(autoDelete);
                queue.setArguments(arguments);
                //3.写硬盘
                if (durable) {
                    diskDataCenter.insertQueue(queue);
                }
                //4.写硬盘
                memoryDataCenter.insertQueue(queue);
                System.out.println("[VirtualHost] 队列创建成功 queueName = " + queueName);
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 队列创建失败 queueName = " + queueName);
            e.printStackTrace();
            return false;
        }
    }

    //删除队列
    public boolean queueDelete(String queueName) {
        queueName = virtualHostName + queueName;
        try {
            synchronized (queueLocker) {
                //1.找到对应的队列
                MSGQueue queue = memoryDataCenter.getQueue(queueName);
                if (queue == null) {
                    throw new MqException("[VirtualHost] 队列不存在！ 无法删除");
                }
                //2.删除硬盘上的数据
                if (queue.isDurable()) {
                    diskDataCenter.deleteQueue(queueName);
                }
                //3.删除内存上的数据
                memoryDataCenter.deleteQueue(queueName);
                System.out.println("[VirtualHost] 队列删除成功！ queueName = " + queueName);
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 队列删除失败！queueName = " + queueName);
            e.printStackTrace();
            return false;
        }
    }


    /**
     * 绑定操作
     *
     * @param queueName
     * @param exchangeName
     * @param bindingKey
     * @return
     */

    //创建绑定
    public boolean queueBind(String queueName, String exchangeName, String bindingKey) {
        queueName = virtualHostName + queueName;
        exchangeName = virtualHostName + exchangeName;
        try {
            synchronized (exchangeLocker) {
                synchronized (queueLocker) {
                    //1.判断当前的绑定是否已经存在
                    Binding existBinding = memoryDataCenter.getBinding(exchangeName, queueName);
                    if (existBinding != null) {
                        throw new MqException("[VirtualHost] binding已经存在！exchangeName= " + exchangeName);
                    }
                    //2.验证bindingKey是否合法
                    if (!router.checkBindingKey(bindingKey)) {
                        throw new MqException("[VirtualHost] bindingKey非法 bindingKey=" + bindingKey);
                    }
                    //3.创建Binding对象
                    Binding binding = new Binding();
                    binding.setExchangeName(exchangeName);
                    binding.setQueueName(queueName);
                    binding.setBindingKey(bindingKey);
                    //4.获取对应的交换机和队列，如果交换机和队列不存在，这样的绑定也无妨创建
                    MSGQueue queue = memoryDataCenter.getQueue(queueName);
                    if (queue == null) {
                        throw new MqException("[VirtualHost]队列不存在！ queueName=" + queueName);
                    }
                    Exchange exchange = memoryDataCenter.getExchange(exchangeName);
                    if (exchange == null) {
                        throw new MqException("[VirtualHost]交换机不存在！ exchangeName = " + exchangeName);
                    }
                    //5.先写硬盘
                    if (queue.isDurable() && exchange.isDurable()) {
                        diskDataCenter.insertBinding(binding);
                    }
                    //6.写入内存
                    memoryDataCenter.insertBinding(binding);
                }
            }
            System.out.println("[VirtualHost] 绑定创建成功！ exchangeName = " + exchangeName);
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 绑定创建失败! exchangeName =" + exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    //删除绑定
    public boolean queueUnbind(String queueName, String exchangeName) {
        queueName = virtualHostName + queueName;
        exchangeName = virtualHostName + exchangeName;
        try {
            synchronized (exchangeLocker) {
                synchronized (queueLocker) {

                    //1.获取binding是否已经存在
                    Binding binding = memoryDataCenter.getBinding(exchangeName, queueName);
                    if (binding == null) {
                        throw new MqException("[VirtualHost] 删除绑定失败！ 绑定不存在！ exchangeName=" + exchangeName + "queueName" + queueName);
                    }
//            //2.获取对应的队列和交换机，看是否存在
//            MSGQueue queue = memoryDataCenter.getQueue(queueName);
//            if (queue == null){
//                throw new MqException("[VirtualHost]对应的队列不存在！queueName="+queueName);
//            }
//            Exchange exchange =memoryDataCenter.getExchange(exchangeName);
//            if (exchange ==null){
//                throw new MqException("[VirtualHost]对应的交换机不存在！exchangeName = "+exchangeName);
//            }
//            //3.删除硬盘上的数据
//            if (queue.isDurable()&&exchange.isDurable()){
//                diskDataCenter.deleteBinding(binding);
//            }
                    //2.物理绑定是否持久化，都尝试从硬盘上删除，就算不存在，删除也没有副作用
                    diskDataCenter.deleteBinding(binding);
                    //4.删除内存上的数据
                    memoryDataCenter.deleteBinding(binding);
                    System.out.println("[VirtualHost] 删除绑定成功！");
                }
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 删除绑定失败！");
            e.printStackTrace();
            return false;
        }
    }

    //发送消息到指定的交换机/队列中
    public boolean basicPublish(String exchangeName, String routingKey, BasicProperties basicProperties, byte[] body) {
        try {
            //1.转换交换机名字
            exchangeName = virtualHostName + exchangeName;
            //2.检查routingKey是否合法
            if (!router.checkRoutingKey(routingKey)) {
                throw new MqException("[VirtualHost] routing非法！routingKey = " + routingKey);
            }
            //3.查找交换机对象
            Exchange exchange = memoryDataCenter.getExchange(exchangeName);
            if (exchange == null) {
                throw new MqException("[VirtualHost] 交换机不存在！ exchangeName = " + exchangeName);
            }
            //4.判定交换机类型
            if (exchange.getType() == ExchangeType.DIRECT) {
                //按照直接交换机的方式来转发
                //以routingKey作为队列的名字，直接把名字写入指定的队列中
                //此时可以无视绑定关系
                String queueName = virtualHostName + routingKey;
                //5.构造消息对象
                Message message = Message.createMessageWithId(routingKey, basicProperties, body);
                //6.查找该队列名对应的对象
                MSGQueue queue = memoryDataCenter.getQueue(queueName);
                if (queue == null) {
                    throw new MqException("[VirtualHost] 队列不存在！ queueName = " + queueName);
                }
                //7.队列存在，直接给队列中写入消息
                sendMessage(queue, message);

            } else {
                //按照fanout和topic的方式来转发
                //5.找到该交换机关联的所有绑定，并遍历这些绑定对象
                ConcurrentHashMap<String, Binding> bindingsMap = memoryDataCenter.getBindings(exchangeName);
                for (Map.Entry<String, Binding> entry : bindingsMap.entrySet()) {
                    //1)获取绑定对象，判断对应的队列是否存在
                    Binding binding = entry.getValue();
                    MSGQueue queue = memoryDataCenter.getQueue(binding.getQueueName());
                    if (queue == null) {
                        //这里可能会有队列的失败，不抛出异常是因为，不希望因为一个队列的失败而影响其他队列的消息的传输
                        System.out.println("[VirtualHost] basicPublish 发送消息时，发现队列不存在！ queueName = " + bindingsMap);
                        continue;
                    }
                    //2)构造消息对象
                    Message message = Message.createMessageWithId(routingKey, basicProperties, body);
                    //3)判断这个消息是否能转发给该队列
                    //fanout 所有绑定的队列都要转发
                    //topic需要判定一下，bindingKey和routingKey是否匹配
                    if (router.route(exchange.getType(), binding, message)) {
                        continue;
                    }
                    //4) 真正转发给队列
                    sendMessage(queue, message);
                }
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 消息发送失败！");
            e.printStackTrace();
            return false;
        }
    }

    private void sendMessage(MSGQueue queue, Message message) throws IOException, MqException, InterruptedException {
        //此处发送消息，就是把消息写入到硬盘 和 内存 上
        int deliverMode = message.getDeliverMode();
        //deleverMode 为1，不持久化；为2，表示持久化
        if (deliverMode == 2) {
            diskDataCenter.sendMessage(queue, message);
        }
        //写入内存
        memoryDataCenter.sendMessage(queue, message);

        //通知消费者可以消费消息
        consumerManager.notifyConsume(queue.getName());
    }

    //订阅消息
    //添加一个队列的订阅者，当队列收到消息之后，就要把消息推送给对应的订阅者
    //consumerTag:消费者身份标识
    //autoAck:消息被消费完成后，应答的方式，true为自动应答，false为手动应答
    //consumer：是一个回调函数，此处类型设定完成函数式接口，这样后续调用basicConsume并且传实参的时候,就可以写作lambda样子
    public boolean basicConsume(String consumerTag, String queueName, boolean autoAck, Consumer consumer) {
        //构造一个ConsumerEnv对象，把这个队列找到，再把这个Consumer对象添加到这个队列中
        queueName = virtualHostName + queueName;
        try{
            consumerManager.addConsumer(consumerTag,queueName,autoAck,consumer);
            System.out.println("[VirtualHost] basicConsume 成功！ queueName ="+queueName);
            return true;
        }catch (Exception e){
            System.out.println("[VirtualHost] basicConsume 失败！ queueName ="+queueName);
            e.printStackTrace();
            return false;
        }
    }

    public boolean basicAck(String queueName,String messageId){
        queueName = virtualHostName + queueName;
        try {
            //1.获取消息和队列
            Message message = memoryDataCenter.getMessage(messageId);
            if (message == null){
                throw new MqException("[VirtualHost] 要确认的消息不存在！ messageId = "+messageId);
            }
            MSGQueue queue = memoryDataCenter.getQueue(queueName);
            if (queue == null){
                throw new MqException("[VirtualHost] 要确认的队列不存在！ queueName = "+ queueName);
            }
            //2.删除硬盘上的数据
            if (message.getDeliverMode() == 2){
                diskDataCenter.deleteMessage(queue,message);
            }
            //3.删除消息中心上的数据
            memoryDataCenter.removeMessage(messageId);
            //4.删除待确认的集合中的数据
            memoryDataCenter.removeWaitMessageAck(queueName,messageId);
            System.out.println("[VirtualHost] basicAck 成功！消息被成功确认！queueName="+ queueName+",messageId = "+messageId);
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] basicAck 失败！消息未被成功确认！queueName="+ queueName+",messageId = "+messageId);
            e.printStackTrace();
            return false;
        }

    }
}
